package com.whut.monitor.netty.handler;

import com.whut.monitor.service.IHbaseService;
import com.whut.monitor.util.ProtocolUtil;
import com.whut.monitor.util.SyncUtil;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

import static com.whut.monitor.util.ProtocolUtil.*;

/**
 * @author WSJ
 * 2017/10/16
 */
@Service("SyncDataHandler")
@ChannelHandler.Sharable
public class SyncDataHandler extends ChannelInboundHandlerAdapter {

    private String msg;

    @Autowired
    IHbaseService hbaseService;


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
        long nowTime = System.currentTimeMillis();
        //判断同步标识
        msg = (String)message;
        String[] msgarr = msg.split("\n");
        if (msgarr.length > 2 && ("HbaseCP".equals(msgarr[1]))){
            //开始解析
            Map<String, String> map = new HashMap<>();
            SyncUtil.msg2Hbase(msg, map);
            saveMap("hbaseSyncMap_NoFix_", map, "");
            for(String key: map.keySet())
            {
                if("ST".equals(key.substring(2, 4))) {
                    map.put(key, waveFixSST(map.get(key)));
                }
                if("FT".equals(key.substring(2, 4))) {
                    map.put(key, waveFixSFT(map.get(key)));
                }
            }
            if (!ProtocolUtil.TestMode) {
                hbaseService.syncData(map);
            }

            String costTime = (System.currentTimeMillis()-nowTime)/1000.0+"s";
            saveString("dataPackage_SyncData_", msg, costTime);
            saveMap("hbaseSyncMap_Fixed_", map, costTime);
        }else {
            // 字符串中第二段无同步标示“HbaseCP”，则传递到ServerInboundHandler
            ctx.fireChannelRead(msg);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        ctx.close();
    }
}
